#pragma once
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "muduo/net/TcpConnection.h"
#include "mq_consumer.hpp"
#include "muduo/proto/codec.h"
#include "mq_threadpool.hpp"
#include "mq_router.hpp"
#include "mq_virhost.hpp"
#include "../mqcommon/mq_proto.pb.h"
namespace mymq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;
    class Channel
    {
    public:
        using Ptr = std::shared_ptr<Channel>;
        Channel(const std::string &cid,
                const muduo::net::TcpConnectionPtr &conn,
                const ProtobufCodecPtr &code,
                const mymq::ConsumerManager::Ptr &cmp,
                const VirtualHost::Ptr &host,
                const Threadpool::Ptr &pool)
            : _cid(cid),
              _conn(conn),
              _code(code),
              _cmp(cmp),
              _host(host),
              _pool(pool)
        {
            DLOG("信道已创建");
        }
        // 声明交换机
        void DectoryExchange(const declareExchangeRequestPtr &req)
        {
            bool ret = _host->DeclareExchange(req->exchange_name(), req->type(), req->durable(), req->auto_delete(), req->args());
            return basicResponse(ret, req->rid(), req->cid());
        }
        // 删除交换机
        void DeleteExchange(const deleteExchangeRequestPtr &req)
        {
            _host->DeleteExchange(req->exchange_name());
            return basicResponse(true, req->rid(), req->cid());
        }
        // 声明队列
        void DeclareQueue(const declareQueueRequestPtr &req)
        {
            // 在声明队列成功时还需要初始化该队列的消费者
            bool ret = _host->DeclareMsgQueue(req->queue_name(),
                                              req->durable(),
                                              req->exclusive(),
                                              req->auto_delete(),
                                              req->args());
            if (ret == false)
                return basicResponse(false, req->rid(), req->cid());
            _cmp->InitQueueConsumer(req->queue_name());
            return basicResponse(ret, req->rid(), req->cid());
        }
        // 删除队列
        void DeleteQueue(const deleteQueueRequestPtr &req)
        {
            // 删除队列时也需要将与其相关的消息，绑定信息,订阅该队列的消费者销毁
            _cmp->DestoryQueueConsumer(req->queue_name());
            _host->DeleteMsgQueue(req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }
        // 绑定
        void QueueBind(const queueBindRequestPtr &req)
        {
            bool ret = _host->Bind(req->exchange_name(), req->queue_name(), req->binding_key());
            return basicResponse(ret, req->rid(), req->cid());
        }
        // 解绑
        void QueueUnBind(const queueUnBindRequestPtr &req)
        {
            bool ret = _host->UnBind(req->exchange_name(), req->queue_name());
            return basicResponse(ret, req->rid(), req->cid());
        }
        // 订阅队列消息
        void BasicConsume(const basicConsumeRequestPtr &req)
        {
            // 先判断该队列是否存在
            bool ret = _host->existsQueue(req->queue_name());
            if (ret == false)
            {
                return basicResponse(false, req->rid(), req->cid());
            }
            // 创建消费者
            auto callback = std::bind(&Channel::CallBack, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
            _consumer = _cmp->Create(req->consumer_tag(), req->queue_name(), req->auto_ack(), callback);
            return basicResponse(true, req->rid(), req->cid());
        }
        // 取消订阅队列消息
        void BasicCancel(const basicCancelRequestPtr &req)
        {
            bool ret = _cmp->Remove(req->consumer_tag(), req->queue_name());
            return basicResponse(ret, req->rid(), req->cid());
        }
        // 发布消息
        void BasicPublish(const basicPublishRequestPtr &req)
        {
            //DLOG("收到请求");
            // 1.先把消息交给交换机
            Exchange::Ptr ep = _host->SelectExchange(req->exchange_name());
            if (ep.get() == nullptr)
            {
                // 没有该交换机
                DLOG("没有该交换机");
                return basicResponse(false, req->rid(), req->cid());
            }
            // 2.交换机再将消息交给与他绑定的队列中
            //      1.获取与该交换机绑定的所有队列
            QueueBindingMap qbmp = _host->ExchangeBindings(ep->name);
            BasicProperties *bp = nullptr;
            std::string routing_key;
            if (req->has_properties())
            {
                bp = req->mutable_properties();
                routing_key = req->properties().routing_key();
            }
            for (auto &qbm : qbmp)
            {
                if (Router::route(ep->type, routing_key, qbm.second->binding_key))
                {
                    // 匹配成功
                    // 将消息添加到指定队列
                    //  3.根据消息的routing_key与队列的binding_key把消息交给对应的队列
                    DLOG("进行消息路由%s",routing_key.c_str());
                    _host->BasicPublish(qbm.first, bp, req->body());
                    // 向线程池中添加一个消息消费任务（向指定队列的订阅者去推送消息--线程池完成）
                    auto task = std::bind(&Channel::Consume, this, qbm.first);
                    _pool->push(task);
                }
            }
            return basicResponse(true, req->rid(), req->cid());
        }
        // 确认消息
        void BasicAck(const basicAckRequestPtr &req)
        {
            _host->BasicAck(req->queue_name(), req->message_id());
            return basicResponse(true, req->rid(), req->cid());
        }

        ~Channel()
        {
            if (_consumer.get() != nullptr)
            {
                DLOG("信道已删除");
                _cmp->Remove(_consumer->_tag, _consumer->_qname);
            }
        }

    private:
        // 消费者的回调处理函数
        void CallBack(const std::string &tag, const BasicProperties *bp, const std::string &body)
        {
            // 针对收到的消息发送给该信道对应的客户端
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_consumer_tag(tag);
            resp.set_body(body);
            if (bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery(bp->delivery());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _code->send(_conn, resp);
        }
        // 消费消息
        void Consume(const std::string &qname)
        {
            // 1.从队列中取出一条消息
            MessagePtr msg = _host->BasicConsume(qname);
            if (msg.get() == nullptr)
            {
                DLOG("从该队列 %s 取出消息为空", qname.c_str());
                return;
            }
            // 2.取出订阅该队列的其中一个消费者
            Consumer::Ptr con = _cmp->Choose(qname);
            if (con.get() == nullptr)
            {
                DLOG("没有消费者订阅该队列 %s", qname.c_str());
                return;
            }
            // 3.调用订阅者对应的回调处理函数进行消息处理
            con->_callback(con->_tag, msg->mutable_payload()->mutable_properties(), msg->payload().body());
            // 4.判断是否为自动确认
            if (con->_auto_ack)
                _host->BasicAck(qname, msg->payload().properties().id());
        }
        // 通用响应
        void basicResponse(bool ok, const std::string &rid, const std::string &cid)
        {
            basicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            _code->send(_conn, resp);
        }

    private:
        std::string _cid;                   // 信道id
        Consumer::Ptr _consumer;            // 信道关联的消费者
        muduo::net::TcpConnectionPtr _conn; // 网络通信协议
        ProtobufCodecPtr _code;             // 协议处理
        VirtualHost::Ptr _host;             // 虚拟机句柄
        mymq::ConsumerManager::Ptr _cmp;    // 消费者管理句柄
        Threadpool::Ptr _pool;              // 线程池
    };
    class ChannelManager
    {
    public:
        using Ptr = std::shared_ptr<ChannelManager>;
        ChannelManager() {}
        // 创建信道
        bool OpenChannel(const std::string &cid,
                           const muduo::net::TcpConnectionPtr &conn,
                           const ProtobufCodecPtr &code,
                           const mymq::ConsumerManager::Ptr &cmp,
                           const VirtualHost::Ptr &host,
                           const Threadpool::Ptr &pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            //先根据信道id查找是否已经存在该信道了
            auto it = _channels.find(cid);
            if(it != _channels.end())
            {
                DLOG("该信道已经存在了");
                return false;
            }
            //创建
            Channel::Ptr cha = std::make_shared<Channel>(cid,conn,code,cmp,host,pool);
            _channels.insert(std::make_pair(cid,cha));
            return true;
        }
        void CloseChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }
        //获取一个信道的信息
        Channel::Ptr GetChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if(it == _channels.end())
            {
                DLOG("不存在该信道 %s",cid.c_str());
                return Channel::Ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::Ptr> _channels;
    };
}